#include <fstream>
#include <random>
#include <thread>

#include <bthread/bthread.h>
#include <glog/logging.h>
#include <gtest/gtest.h>

#include "hawking/indexlib/index_config.pb.h"
#include "doc/document_creator.h"
#include "index/inverted_index/inverted_mem_indexer/inverted_mem_indexer.h"
#include "util/file_wrapper/file_wrapper.h"

using namespace hawking::indexlib;
std::uniform_int_distribution<int32_t> dis(1, 100);
std::default_random_engine e{std::random_device{}()};

DEFINE_uint32(write_count, 0, "for dump test");
std::unordered_map<DocId, DocId> old2new_map;

void* TermWrite(void* arg) {
    InvertedMemIndexer* inverted_mem_indexer = reinterpret_cast<InvertedMemIndexer*>(arg);

    // makeup doc creator
    std::shared_ptr<IndexlibIndexConfig> fake_index_config =
        std::make_shared<IndexlibIndexConfig>();
    auto* fake_field = fake_index_config->add_fields();
    fake_field->set_field_id(0);
    fake_field->set_field_name("fake_inverted_field");
    fake_field->set_field_type(hawking::util::ColumnType::COLUMN_INT64);

    auto* fake_invterd_field = fake_index_config->add_inverted_configs();
    fake_invterd_field->set_field_id(0);
    fake_invterd_field->set_field_name("fake_inverted_field");

    fake_index_config->set_pk_field_id(0);

    DocumentCreator doc_creator(fake_index_config);

    bool write_limite = false;
    if (FLAGS_write_count) {
        write_limite = true;
    }

    DocId docid = 0;
    while (1) {
        hawking::util::IndexlibMessage message;
        message.set_cmd(hawking::util::IndexOp::INDEX_RT_ADD);
        auto* column = message.add_columns();
        column->set_column_name("fake_inverted_field");
        int32_t fake_term = dis(e);
        column->mutable_column_value()->set_int_value(fake_term);
        // LOG(INFO) << "add one for " << fake_term;
        std::shared_ptr<NormalDocument> doc = doc_creator.CreateNormalDocument(message, 0);
        doc->SetDocId(docid);
        old2new_map.emplace(docid, docid);
        ++docid;

        inverted_mem_indexer->AddDocument(doc.get());
        bthread_usleep(10 * 1000);

        if (write_limite && docid == FLAGS_write_count) {
            break;
        }
    }

    return nullptr;
}

void* TermRead(void* arg) {
    InvertedMemIndexer* inverted_mem_indexer = reinterpret_cast<InvertedMemIndexer*>(arg);

    while (1) {
        int32_t fake_term = dis(e);
        std::string str = std::to_string(fake_term);
        uint64_t term_key = HashAlgorithm::HashString64(str.data(), str.length());

        hawking::util::ArenaWrapper arena_wrapper;
        SegmentContext segment_ctx(arena_wrapper);
        inverted_mem_indexer->Lookup(term_key, 2000, &segment_ctx);
        // LOG(INFO) << "read docs " << segment_ctx.RelateDocCount() << " of " << fake_term;
        bthread_usleep(10 * 1000);
    }
    
    return nullptr;
}

TEST(inverted_test, read_and_write) {
    InvertedMemIndexer inverted_mem_indexer;
    inverted_mem_indexer.Init(0);

    bthread_t writer;
    bthread_start_background(&writer, nullptr, TermWrite, &inverted_mem_indexer);

    std::vector<bthread_t> readers;
    readers.reserve(128);
    for (size_t idx = 0; idx < readers.capacity(); ++idx) {
        readers.emplace_back();
        bthread_start_background(&readers.back(), nullptr, TermRead, &inverted_mem_indexer);
    }

    bthread_join(writer, nullptr);
    for (auto reader : readers) {
        bthread_join(reader, nullptr);
    }
}

TEST(inverted_test, dump) {
    InvertedMemIndexer inverted_mem_indexer;
    inverted_mem_indexer.Init(0);

    bthread_t writer;
    bthread_start_background(&writer, nullptr, TermWrite, &inverted_mem_indexer);
    bthread_join(writer, nullptr);

    LOG(INFO) << "begin dump";
    FslibWrapper::RmDir("/tmp/inverted_test");
    FslibWrapper::Mkdir("/tmp/inverted_test");
    inverted_mem_indexer.Dump("/tmp/inverted_test", old2new_map);
}

int main(int argc, char* argv[]) {
    ::testing::InitGoogleTest(&argc, argv);
    ::google::ParseCommandLineFlags(&argc, &argv, true);
    srand(static_cast<unsigned>(time(0)));
    return RUN_ALL_TESTS();
}
